#include "executor.h"
#include "../storage/storage_engine.h"
#include "../planner/query_planner.h"
#include <string.h>
#include <stdlib.h>
#include <assert.h>
#include <math.h>
#include <inttypes.h>

// Static function declarations
static executor_state_t* create_scan_executor(executor_context_t* context, plan_node_t* plan);
static executor_state_t* create_index_scan_executor(executor_context_t* context, plan_node_t* plan);
static executor_state_t* create_filter_executor(executor_context_t* context, plan_node_t* plan);
static executor_state_t* create_project_executor(executor_context_t* context, plan_node_t* plan);
static executor_state_t* create_join_executor(executor_context_t* context, plan_node_t* plan);
static executor_state_t* create_aggregate_executor(executor_context_t* context, plan_node_t* plan);
static executor_state_t* create_sort_executor(executor_context_t* context, plan_node_t* plan);
static executor_state_t* create_limit_executor(executor_context_t* context, plan_node_t* plan);

// Scan executor functions
static exec_result_t scan_open(executor_state_t* state);
static exec_result_t scan_next(executor_state_t* state, tuple_t** tuple);
static void scan_close(executor_state_t* state);
static void scan_rescan(executor_state_t* state);

// Index scan executor functions
static exec_result_t index_scan_open(executor_state_t* state);
static exec_result_t index_scan_next(executor_state_t* state, tuple_t** tuple);
static void index_scan_close(executor_state_t* state);
static void index_scan_rescan(executor_state_t* state);

// Filter executor functions
static exec_result_t filter_open(executor_state_t* state);
static exec_result_t filter_next(executor_state_t* state, tuple_t** tuple);
static void filter_close(executor_state_t* state);
static void filter_rescan(executor_state_t* state);

// Project executor functions
static exec_result_t project_open(executor_state_t* state);
static exec_result_t project_next(executor_state_t* state, tuple_t** tuple);
static void project_close(executor_state_t* state);
static void project_rescan(executor_state_t* state);

// Join executor functions
static exec_result_t join_open(executor_state_t* state);
static exec_result_t join_next(executor_state_t* state, tuple_t** tuple);
static void join_close(executor_state_t* state);
static void join_rescan(executor_state_t* state);

// Sort executor functions
static exec_result_t sort_open(executor_state_t* state);
static exec_result_t sort_next(executor_state_t* state, tuple_t** tuple);
static void sort_close(executor_state_t* state);
static void sort_rescan(executor_state_t* state);

// Sort utility types and functions
typedef struct {
    expression_t** sort_keys;
    bool* ascending;
    uint32_t key_count;
} sort_context_t;

static int tuple_compare_wrapper(const void* a, const void* b, void* context);
static void quicksort_tuples(tuple_t** tuples, int low, int high, sort_context_t* sort_ctx);
static int partition_tuples(tuple_t** tuples, int low, int high, sort_context_t* sort_ctx);

// Aggregate executor functions
static exec_result_t aggregate_open(executor_state_t* state);
static exec_result_t aggregate_next(executor_state_t* state, tuple_t** tuple);
static void aggregate_close(executor_state_t* state);
static void aggregate_rescan(executor_state_t* state);

// Limit executor functions
static exec_result_t limit_open(executor_state_t* state);
static exec_result_t limit_next(executor_state_t* state, tuple_t** tuple);
static void limit_close(executor_state_t* state);
static void limit_rescan(executor_state_t* state);

// Executor context management
executor_context_t* executor_create_context(gudb* db, gudb_txn* txn, memory_manager_t* mm) {
    executor_context_t* context = malloc(sizeof(executor_context_t));
    if (!context) {
        return NULL;
    }
    
    memset(context, 0, sizeof(executor_context_t));
    
    context->db = db;
    context->transaction = txn;
    context->memory_manager = mm;
    
    // Set default configuration
    context->work_mem_kb = 4096;        // 4MB working memory
    context->temp_buffers = 8;
    context->enable_hashjoin = true;
    context->enable_mergejoin = true;
    context->enable_nestloop = true;
    
    return context;
}

void executor_destroy_context(executor_context_t* context) {
    if (!context) return;
    
    if (context->error_message) {
        free(context->error_message);
    }
    
    if (context->parameters) {
        for (uint32_t i = 0; i < context->parameter_count; i++) {
            value_destroy(&context->parameters[i]);
        }
        free(context->parameters);
    }
    
    // Clean up correlation context
    if (context->correlation_values) {
        for (uint32_t i = 0; i < context->correlation_count; i++) {
            value_destroy(&context->correlation_values[i]);
        }
        free(context->correlation_values);
    }
    
    if (context->correlation_names) {
        for (uint32_t i = 0; i < context->correlation_count; i++) {
            free(context->correlation_names[i]);
        }
        free(context->correlation_names);
    }
    
    free(context);
}

// Executor tree creation
executor_state_t* executor_create_tree(executor_context_t* context, plan_node_t* plan) {
    if (!context || !plan) {
        return NULL;
    }
    
    switch (plan->type) {
        case PLAN_SCAN:
            return create_scan_executor(context, plan);
        case PLAN_INDEX_SCAN:
            return create_index_scan_executor(context, plan);
        case PLAN_FILTER:
            return create_filter_executor(context, plan);
        case PLAN_PROJECT:
            return create_project_executor(context, plan);
        case PLAN_JOIN:
            return create_join_executor(context, plan);
        case PLAN_AGGREGATE:
            return create_aggregate_executor(context, plan);
        case PLAN_SORT:
            return create_sort_executor(context, plan);
        case PLAN_LIMIT:
            return create_limit_executor(context, plan);
        default:
            executor_set_error(context, -1, "Unsupported plan node type");
            return NULL;
    }
}

void executor_destroy_tree(executor_state_t* executor) {
    if (!executor) return;
    
    // Close if still open
    if (executor->is_open) {
        executor->close(executor);
    }
    
    // Destroy children recursively
    if (executor->children) {
        for (uint32_t i = 0; i < executor->child_count; i++) {
            executor_destroy_tree(executor->children[i]);
        }
        free(executor->children);
    }
    
    // Destroy private state
    if (executor->private_state) {
        free(executor->private_state);
    }
    
    free(executor);
}

// Main execution control
exec_result_t executor_open(executor_state_t* executor) {
    if (!executor) return EXEC_ERROR;
    return executor->open(executor);
}

exec_result_t executor_next(executor_state_t* executor, tuple_t** tuple) {
    if (!executor) return EXEC_ERROR;
    return executor->next(executor, tuple);
}

void executor_close(executor_state_t* executor) {
    if (!executor) return;
    executor->close(executor);
}

void executor_rescan(executor_state_t* executor) {
    if (!executor) return;
    executor->rescan(executor);
}

// High-level execution functions
exec_result_t executor_run_query(executor_context_t* context, plan_node_t* plan, 
                               tuple_t*** results, uint32_t* result_count) {
    if (!context || !plan || !results || !result_count) {
        return EXEC_ERROR;
    }
    
    *results = NULL;
    *result_count = 0;
    
    executor_state_t* executor = executor_create_tree(context, plan);
    if (!executor) {
        return EXEC_ERROR;
    }
    
    exec_result_t result = executor_open(executor);
    if (result != EXEC_SUCCESS) {
        executor_destroy_tree(executor);
        return result;
    }
    
    // Collect all results
    tuple_t** result_array = NULL;
    uint32_t capacity = 0;
    uint32_t count = 0;
    
    tuple_t* tuple;
    while ((result = executor_next(executor, &tuple)) == EXEC_SUCCESS) {
        if (count >= capacity) {
            capacity = capacity == 0 ? 16 : capacity * 2;
            result_array = realloc(result_array, capacity * sizeof(tuple_t*));
            if (!result_array) {
                result = EXEC_ERROR;
                break;
            }
        }
        
        result_array[count++] = tuple_copy(tuple);
        context->total_tuples_processed++;
    }
    
    executor_close(executor);
    executor_destroy_tree(executor);
    
    if (result == EXEC_EOF) {
        *results = result_array;
        *result_count = count;
        return EXEC_SUCCESS;
    } else {
        // Error occurred, clean up
        for (uint32_t i = 0; i < count; i++) {
            tuple_destroy(result_array[i]);
        }
        free(result_array);
        return result;
    }
}

exec_result_t executor_run_dml(executor_context_t* context, plan_node_t* plan, 
                              uint64_t* affected_rows) {
    if (!context || !plan || !affected_rows) {
        return EXEC_ERROR;
    }
    
    *affected_rows = 0;
    
    executor_state_t* executor = executor_create_tree(context, plan);
    if (!executor) {
        return EXEC_ERROR;
    }
    
    exec_result_t result = executor_open(executor);
    if (result != EXEC_SUCCESS) {
        executor_destroy_tree(executor);
        return result;
    }
    
    tuple_t* tuple;
    while ((result = executor_next(executor, &tuple)) == EXEC_SUCCESS) {
        (*affected_rows)++;
        context->total_tuples_processed++;
    }
    
    executor_close(executor);
    executor_destroy_tree(executor);
    
    return (result == EXEC_EOF) ? EXEC_SUCCESS : result;
}

// Tuple management
tuple_t* tuple_create(uint32_t column_count) {
    tuple_t* tuple = malloc(sizeof(tuple_t));
    if (!tuple) return NULL;
    
    tuple->slot = tuple_slot_create(column_count);
    if (!tuple->slot) {
        free(tuple);
        return NULL;
    }
    
    tuple->tuple_id = 0;
    tuple->table_oid = NULL;
    tuple->is_heap_tuple = true;
    
    return tuple;
}

tuple_t* tuple_copy(tuple_t* source) {
    if (!source) return NULL;
    
    tuple_t* copy = malloc(sizeof(tuple_t));
    if (!copy) return NULL;
    
    copy->slot = tuple_slot_create(source->slot->column_count);
    if (!copy->slot) {
        free(copy);
        return NULL;
    }
    
    // Copy values
    for (uint32_t i = 0; i < source->slot->column_count; i++) {
        copy->slot->values[i] = value_copy(&source->slot->values[i]);
        copy->slot->is_null[i] = source->slot->is_null[i];
    }
    
    copy->tuple_id = source->tuple_id;
    copy->table_oid = source->table_oid;
    copy->is_heap_tuple = source->is_heap_tuple;
    
    return copy;
}

void tuple_destroy(tuple_t* tuple) {
    if (!tuple) return;
    
    tuple_slot_destroy(tuple->slot);
    free(tuple);
}

tuple_slot_t* tuple_slot_create(uint32_t column_count) {
    tuple_slot_t* slot = malloc(sizeof(tuple_slot_t));
    if (!slot) return NULL;
    
    slot->values = calloc(column_count, sizeof(value_t));
    slot->is_null = calloc(column_count, sizeof(bool));
    
    if (!slot->values || !slot->is_null) {
        free(slot->values);
        free(slot->is_null);
        free(slot);
        return NULL;
    }
    
    slot->column_count = column_count;
    slot->ref_count = 1;
    
    return slot;
}

void tuple_slot_destroy(tuple_slot_t* slot) {
    if (!slot) return;
    
    if (--slot->ref_count > 0) {
        return;
    }
    
    if (slot->values) {
        for (uint32_t i = 0; i < slot->column_count; i++) {
            value_destroy(&slot->values[i]);
        }
        free(slot->values);
    }
    
    free(slot->is_null);
    free(slot);
}

// Value management
value_t value_make_null(void) {
    value_t value;
    value.type = VALUE_NULL;
    value.is_null = true;
    memset(&value.data, 0, sizeof(value.data));
    return value;
}

value_t value_make_bool(bool val) {
    value_t value;
    value.type = VALUE_BOOL;
    value.is_null = false;
    value.data.bool_val = val;
    return value;
}

value_t value_make_int32(int32_t val) {
    value_t value;
    value.type = VALUE_INT32;
    value.is_null = false;
    value.data.int32_val = val;
    return value;
}

value_t value_make_int64(int64_t val) {
    value_t value;
    value.type = VALUE_INT64;
    value.is_null = false;
    value.data.int64_val = val;
    return value;
}

value_t value_make_string(const char* str) {
    value_t value;
    value.type = VALUE_STRING;
    value.is_null = (str == NULL);
    
    if (str) {
        uint32_t len = strlen(str);
        value.data.string_val.data = malloc(len + 1);
        if (value.data.string_val.data) {
            strcpy(value.data.string_val.data, str);
            value.data.string_val.length = len;
        } else {
            value.is_null = true;
            value.data.string_val.length = 0;
        }
    } else {
        value.data.string_val.data = NULL;
        value.data.string_val.length = 0;
    }
    
    return value;
}

value_t value_make_float(float val) {
    value_t value;
    value.type = VALUE_FLOAT;
    value.is_null = false;
    value.data.float_val = val;
    return value;
}

value_t value_make_double(double val) {
    value_t value;
    value.type = VALUE_DOUBLE;
    value.is_null = false;
    value.data.double_val = val;
    return value;
}

value_t value_copy(const value_t* source) {
    value_t copy = *source;
    
    if (!source->is_null && source->type == VALUE_STRING && source->data.string_val.data) {
        copy.data.string_val.data = malloc(source->data.string_val.length + 1);
        if (copy.data.string_val.data) {
            memcpy(copy.data.string_val.data, source->data.string_val.data, 
                   source->data.string_val.length + 1);
        } else {
            copy.is_null = true;
            copy.data.string_val.length = 0;
        }
    } else if (!source->is_null && source->type == VALUE_BLOB && source->data.blob_val.data) {
        copy.data.blob_val.data = malloc(source->data.blob_val.length);
        if (copy.data.blob_val.data) {
            memcpy(copy.data.blob_val.data, source->data.blob_val.data, 
                   source->data.blob_val.length);
        } else {
            copy.is_null = true;
            copy.data.blob_val.length = 0;
        }
    }
    
    return copy;
}

void value_destroy(value_t* value) {
    if (!value || value->is_null) return;
    
    if (value->type == VALUE_STRING && value->data.string_val.data) {
        free(value->data.string_val.data);
        value->data.string_val.data = NULL;
    } else if (value->type == VALUE_BLOB && value->data.blob_val.data) {
        free(value->data.blob_val.data);
        value->data.blob_val.data = NULL;
    }
    
    value->is_null = true;
}

bool value_equals(const value_t* a, const value_t* b) {
    if (a->is_null && b->is_null) return true;
    if (a->is_null || b->is_null) return false;
    if (a->type != b->type) return false;
    
    switch (a->type) {
        case VALUE_BOOL:
            return a->data.bool_val == b->data.bool_val;
        case VALUE_INT32:
            return a->data.int32_val == b->data.int32_val;
        case VALUE_INT64:
            return a->data.int64_val == b->data.int64_val;
        case VALUE_FLOAT:
            return fabs(a->data.float_val - b->data.float_val) < 1e-6;
        case VALUE_DOUBLE:
            return fabs(a->data.double_val - b->data.double_val) < 1e-15;
        case VALUE_STRING:
            if (a->data.string_val.length != b->data.string_val.length) return false;
            return strncmp(a->data.string_val.data, b->data.string_val.data, 
                          a->data.string_val.length) == 0;
        case VALUE_TIMESTAMP:
            return a->data.timestamp_val == b->data.timestamp_val;
        default:
            return false;
    }
}

int value_compare(const value_t* a, const value_t* b) {
    if (a->is_null && b->is_null) return 0;
    if (a->is_null) return -1;
    if (b->is_null) return 1;
    
    if (a->type != b->type) {
        return (int)a->type - (int)b->type;
    }
    
    switch (a->type) {
        case VALUE_BOOL:
            return (int)a->data.bool_val - (int)b->data.bool_val;
        case VALUE_INT32:
            if (a->data.int32_val < b->data.int32_val) return -1;
            if (a->data.int32_val > b->data.int32_val) return 1;
            return 0;
        case VALUE_INT64:
            if (a->data.int64_val < b->data.int64_val) return -1;
            if (a->data.int64_val > b->data.int64_val) return 1;
            return 0;
        case VALUE_FLOAT:
            if (a->data.float_val < b->data.float_val) return -1;
            if (a->data.float_val > b->data.float_val) return 1;
            return 0;
        case VALUE_DOUBLE:
            if (a->data.double_val < b->data.double_val) return -1;
            if (a->data.double_val > b->data.double_val) return 1;
            return 0;
        case VALUE_STRING:
            return strncmp(a->data.string_val.data, b->data.string_val.data,
                          (a->data.string_val.length < b->data.string_val.length) ?
                          a->data.string_val.length : b->data.string_val.length);
        case VALUE_TIMESTAMP:
            if (a->data.timestamp_val < b->data.timestamp_val) return -1;
            if (a->data.timestamp_val > b->data.timestamp_val) return 1;
            return 0;
        default:
            return 0;
    }
}

// Expression evaluation (complete implementation)
exec_result_t evaluate_expression(expression_t* expr, tuple_t* tuple, 
                                 executor_context_t* context, value_t* result) {
    if (!expr || !result) {
        return EXEC_ERROR;
    }
    
    switch (expr->type) {
        case EXPR_LITERAL: {
            // Handle literal values
            if (expr->data.literal.value == NULL) {
                *result = value_make_null();
                return EXEC_SUCCESS;
            }
            
            // Parse literal based on data type
            switch (expr->data_type) {
                case DATA_TYPE_INT:
                case DATA_TYPE_BIGINT:
                case DATA_TYPE_TINYINT: {
                    int64_t val = strtoll(expr->data.literal.value, NULL, 10);
                    *result = value_make_int64(val);
                    break;
                }
                case DATA_TYPE_FLOAT:
                case DATA_TYPE_DOUBLE: {
                    double val = strtod(expr->data.literal.value, NULL);
                    *result = value_make_double(val);
                    break;
                }
                case DATA_TYPE_VARCHAR:
                case DATA_TYPE_CHAR:
                case DATA_TYPE_TEXT:
                default: {
                    *result = value_make_string(expr->data.literal.value);
                    break;
                }
            }
            return EXEC_SUCCESS;
        }
        
        case EXPR_COLUMN: {
            // Check if this is a correlated reference first
            if (context->correlation_count > 0 && context->correlation_names) {
                // Build full column reference
                char* full_ref = NULL;
                if (expr->data.column.table_name) {
                    size_t len = strlen(expr->data.column.table_name) + strlen(expr->data.column.column_name) + 2;
                    full_ref = malloc(len);
                    if (full_ref) {
                        snprintf(full_ref, len, "%s.%s", expr->data.column.table_name, expr->data.column.column_name);
                    }
                } else {
                    full_ref = strdup(expr->data.column.column_name);
                }
                
                if (full_ref) {
                    // Check if this column is in the correlation context
                    for (uint32_t i = 0; i < context->correlation_count; i++) {
                        if (context->correlation_names[i] && strcmp(context->correlation_names[i], full_ref) == 0) {
                            // Found correlated reference - return the correlated value
                            *result = value_copy(&context->correlation_values[i]);
                            free(full_ref);
                            return EXEC_SUCCESS;
                        }
                    }
                    free(full_ref);
                }
            }
            
            // Get column value from tuple
            if (!tuple || !tuple->slot) {
                *result = value_make_null();
                return EXEC_SUCCESS;
            }
            
            // For now, simple column lookup by name
            // In a real implementation, this would use column indexes
            // This is a simplified version that assumes column names are stored somewhere
            *result = value_make_null(); // Placeholder
            return EXEC_SUCCESS;
        }
        
        case EXPR_BINARY_OP: {
            value_t left_val, right_val;
            exec_result_t left_result = evaluate_expression(expr->data.binary_op.left, tuple, context, &left_val);
            if (left_result != EXEC_SUCCESS) {
                return left_result;
            }
            
            exec_result_t right_result = evaluate_expression(expr->data.binary_op.right, tuple, context, &right_val);
            if (right_result != EXEC_SUCCESS) {
                value_destroy(&left_val);
                return right_result;
            }
            
            // Handle NULL values
            if (left_val.is_null || right_val.is_null) {
                value_destroy(&left_val);
                value_destroy(&right_val);
                *result = value_make_null();
                return EXEC_SUCCESS;
            }
            
            // Perform binary operation
            exec_result_t op_result = evaluate_binary_operation(expr->data.binary_op.operator, 
                                                               &left_val, &right_val, result);
            
            value_destroy(&left_val);
            value_destroy(&right_val);
            return op_result;
        }
        
        case EXPR_UNARY_OP: {
            value_t operand_val;
            exec_result_t operand_result = evaluate_expression(expr->data.unary_op.operand, tuple, context, &operand_val);
            if (operand_result != EXEC_SUCCESS) {
                return operand_result;
            }
            
            if (operand_val.is_null) {
                value_destroy(&operand_val);
                *result = value_make_null();
                return EXEC_SUCCESS;
            }
            
            exec_result_t op_result = evaluate_unary_operation(expr->data.unary_op.operator, 
                                                              &operand_val, result);
            
            value_destroy(&operand_val);
            return op_result;
        }
        
        case EXPR_FUNCTION_CALL: {
            // Handle aggregate and scalar functions
            return evaluate_function_call(expr, tuple, context, result);
        }
        
        case EXPR_STAR: {
            // SELECT * - this should be handled at a higher level
            *result = value_make_null();
            return EXEC_SUCCESS;
        }
        
        case EXPR_LIST: {
            // Expression lists (used in IN clauses, etc.)
            *result = value_make_null();
            return EXEC_SUCCESS;
        }
        
        case EXPR_SUBQUERY: {
            // Execute the subquery and return appropriate result
            if (!expr->data.subquery.query) {
                *result = value_make_null();
                return EXEC_ERROR;
            }
            
            // Execute the subquery using a separate executor context
            statement_t* subquery_stmt = expr->data.subquery.query;
            
            // Handle correlated subqueries
            if (expr->data.subquery.is_correlated) {
                // Create correlation context from outer query values
                if (expr->data.subquery.correlation_count > 0) {
                    context->correlation_values = calloc(expr->data.subquery.correlation_count, sizeof(value_t));
                    context->correlation_names = calloc(expr->data.subquery.correlation_count, sizeof(char*));
                    context->correlation_count = expr->data.subquery.correlation_count;
                    
                    if (!context->correlation_values || !context->correlation_names) {
                        free(context->correlation_values);
                        free(context->correlation_names);
                        context->correlation_values = NULL;
                        context->correlation_names = NULL;
                        context->correlation_count = 0;
                        *result = value_make_null();
                        return EXEC_ERROR;
                    }
                    
                    // Copy correlation reference names
                    for (uint32_t i = 0; i < expr->data.subquery.correlation_count; i++) {
                        context->correlation_names[i] = strdup(expr->data.subquery.correlation_refs[i]);
                        if (!context->correlation_names[i]) {
                            // Clean up on allocation failure
                            for (uint32_t j = 0; j < i; j++) {
                                free(context->correlation_names[j]);
                            }
                            free(context->correlation_values);
                            free(context->correlation_names);
                            context->correlation_values = NULL;
                            context->correlation_names = NULL;
                            context->correlation_count = 0;
                            *result = value_make_null();
                            return EXEC_ERROR;
                        }
                    }
                    
                    // The actual correlation values will be set by the outer query execution
                    // For now, we'll set them to NULL and they'll be populated during execution
                    for (uint32_t i = 0; i < expr->data.subquery.correlation_count; i++) {
                        context->correlation_values[i] = value_make_null();
                    }
                }
            }
            
            // Execute the subquery as a SELECT statement
            if (subquery_stmt->type == STMT_SELECT) {
                // Create a plan for the subquery
                query_planner_t* planner = planner_create();
                if (!planner) {
                    *result = value_make_null();
                    return EXEC_ERROR;
                }
                
                plan_node_t* subquery_plan = planner_create_plan(planner, subquery_stmt);
                planner_destroy(planner);
                
                if (!subquery_plan) {
                    *result = value_make_null();
                    return EXEC_ERROR;
                }
                
                // Create executor for subquery
                executor_state_t* subquery_executor = executor_create_tree(context, subquery_plan);
                if (!subquery_executor) {
                    plan_destroy(subquery_plan);
                    *result = value_make_null();
                    return EXEC_ERROR;
                }
                
                // Execute subquery and get first result
                exec_result_t open_result = executor_open(subquery_executor);
                if (open_result != EXEC_SUCCESS) {
                    executor_destroy_tree(subquery_executor);
                    plan_destroy(subquery_plan);
                    *result = value_make_null();
                    return EXEC_ERROR;
                }
                
                tuple_t* subquery_tuple = NULL;
                exec_result_t next_result = executor_next(subquery_executor, &subquery_tuple);
                
                if (next_result == EXEC_SUCCESS && subquery_tuple && subquery_tuple->slot) {
                    // Scalar subquery - return first column of first row
                    if (subquery_tuple->slot->column_count > 0) {
                        *result = value_copy(&subquery_tuple->slot->values[0]);
                    } else {
                        *result = value_make_null();
                    }
                } else if (next_result == EXEC_EOF) {
                    // No results from subquery - return NULL
                    *result = value_make_null();
                } else {
                    // Error in execution
                    *result = value_make_null();
                    executor_close(subquery_executor);
                    executor_destroy_tree(subquery_executor);
                    plan_destroy(subquery_plan);
                    return EXEC_ERROR;
                }
                
                // Clean up
                if (subquery_tuple) {
                    tuple_destroy(subquery_tuple);
                }
                executor_close(subquery_executor);
                executor_destroy_tree(subquery_executor);
                plan_destroy(subquery_plan);
                
                // Clean up correlation context
                if (expr->data.subquery.is_correlated && context->correlation_count > 0) {
                    if (context->correlation_values) {
                        for (uint32_t i = 0; i < context->correlation_count; i++) {
                            value_free(&context->correlation_values[i]);
                        }
                        free(context->correlation_values);
                        context->correlation_values = NULL;
                    }
                    if (context->correlation_names) {
                        for (uint32_t i = 0; i < context->correlation_count; i++) {
                            free(context->correlation_names[i]);
                        }
                        free(context->correlation_names);
                        context->correlation_names = NULL;
                    }
                    context->correlation_count = 0;
                }
                
                return EXEC_SUCCESS;
            } else {
                // Non-SELECT subqueries not supported yet
                *result = value_make_null();
                return EXEC_ERROR;
            }
        }
        
        default:
            *result = value_make_null();
            return EXEC_ERROR;
    }
}

// Helper function for binary operations
exec_result_t evaluate_binary_operation(token_type_t operator, const value_t* left_val, 
                                       const value_t* right_val, value_t* result) {
    // Comparison operations
    if (operator == TOKEN_EQUAL || operator == TOKEN_NOT_EQUAL || 
        operator == TOKEN_LESS_THAN || operator == TOKEN_LESS_EQUAL ||
        operator == TOKEN_GREATER_THAN || operator == TOKEN_GREATER_EQUAL) {
        
        int comparison = value_compare(left_val, right_val);
        bool comparison_result = false;
        
        switch (operator) {
            case TOKEN_EQUAL: comparison_result = (comparison == 0); break;
            case TOKEN_NOT_EQUAL: comparison_result = (comparison != 0); break;
            case TOKEN_LESS_THAN: comparison_result = (comparison < 0); break;
            case TOKEN_LESS_EQUAL: comparison_result = (comparison <= 0); break;
            case TOKEN_GREATER_THAN: comparison_result = (comparison > 0); break;
            case TOKEN_GREATER_EQUAL: comparison_result = (comparison >= 0); break;
            default: return EXEC_ERROR;
        }
        
        *result = value_make_bool(comparison_result);
        return EXEC_SUCCESS;
    }
    
    // Arithmetic operations
    if (operator == TOKEN_PLUS || operator == TOKEN_MINUS || 
        operator == TOKEN_MULTIPLY || operator == TOKEN_DIVIDE) {
        
        // Convert values to numeric types
        if (left_val->type == VALUE_INT32 || left_val->type == VALUE_INT64) {
            int64_t left_int = (left_val->type == VALUE_INT32) ? 
                              left_val->data.int32_val : left_val->data.int64_val;
            
            if (right_val->type == VALUE_INT32 || right_val->type == VALUE_INT64) {
                int64_t right_int = (right_val->type == VALUE_INT32) ? 
                                   right_val->data.int32_val : right_val->data.int64_val;
                
                int64_t result_int;
                switch (operator) {
                    case TOKEN_PLUS: result_int = left_int + right_int; break;
                    case TOKEN_MINUS: result_int = left_int - right_int; break;
                    case TOKEN_MULTIPLY: result_int = left_int * right_int; break;
                    case TOKEN_DIVIDE: 
                        if (right_int == 0) return EXEC_ERROR;
                        result_int = left_int / right_int; 
                        break;
                    default: return EXEC_ERROR;
                }
                *result = value_make_int64(result_int);
                return EXEC_SUCCESS;
            }
        }
        
        // Handle double arithmetic
        if (left_val->type == VALUE_DOUBLE || left_val->type == VALUE_FLOAT ||
            right_val->type == VALUE_DOUBLE || right_val->type == VALUE_FLOAT) {
            
            double left_double = (left_val->type == VALUE_DOUBLE) ? left_val->data.double_val :
                                (left_val->type == VALUE_FLOAT) ? left_val->data.float_val :
                                (left_val->type == VALUE_INT32) ? left_val->data.int32_val :
                                left_val->data.int64_val;
            
            double right_double = (right_val->type == VALUE_DOUBLE) ? right_val->data.double_val :
                                 (right_val->type == VALUE_FLOAT) ? right_val->data.float_val :
                                 (right_val->type == VALUE_INT32) ? right_val->data.int32_val :
                                 right_val->data.int64_val;
            
            double result_double;
            switch (operator) {
                case TOKEN_PLUS: result_double = left_double + right_double; break;
                case TOKEN_MINUS: result_double = left_double - right_double; break;
                case TOKEN_MULTIPLY: result_double = left_double * right_double; break;
                case TOKEN_DIVIDE: 
                    if (right_double == 0.0) return EXEC_ERROR;
                    result_double = left_double / right_double; 
                    break;
                default: return EXEC_ERROR;
            }
            *result = value_make_double(result_double);
            return EXEC_SUCCESS;
        }
    }
    
    // Logical operations
    if (operator == TOKEN_AND || operator == TOKEN_OR) {
        if (left_val->type != VALUE_BOOL || right_val->type != VALUE_BOOL) {
            return EXEC_ERROR;
        }
        
        bool result_bool;
        switch (operator) {
            case TOKEN_AND: result_bool = left_val->data.bool_val && right_val->data.bool_val; break;
            case TOKEN_OR: result_bool = left_val->data.bool_val || right_val->data.bool_val; break;
            default: return EXEC_ERROR;
        }
        *result = value_make_bool(result_bool);
        return EXEC_SUCCESS;
    }
    
    *result = value_make_null();
    return EXEC_ERROR;
}

// Helper function for unary operations
exec_result_t evaluate_unary_operation(token_type_t operator, const value_t* operand_val, 
                                      value_t* result) {
    switch (operator) {
        case TOKEN_NOT:
            if (operand_val->type != VALUE_BOOL) {
                return EXEC_ERROR;
            }
            *result = value_make_bool(!operand_val->data.bool_val);
            return EXEC_SUCCESS;
            
        case TOKEN_MINUS:
            if (operand_val->type == VALUE_INT32) {
                *result = value_make_int32(-operand_val->data.int32_val);
                return EXEC_SUCCESS;
            } else if (operand_val->type == VALUE_INT64) {
                *result = value_make_int64(-operand_val->data.int64_val);
                return EXEC_SUCCESS;
            } else if (operand_val->type == VALUE_FLOAT) {
                *result = value_make_float(-operand_val->data.float_val);
                return EXEC_SUCCESS;
            } else if (operand_val->type == VALUE_DOUBLE) {
                *result = value_make_double(-operand_val->data.double_val);
                return EXEC_SUCCESS;
            }
            return EXEC_ERROR;
            
        case TOKEN_PLUS:
            // Unary plus - just return the value as-is
            *result = value_copy(operand_val);
            return EXEC_SUCCESS;
            
        default:
            return EXEC_ERROR;
    }
}

// Helper function for function call evaluation
exec_result_t evaluate_function_call(expression_t* expr, tuple_t* tuple, 
                                    executor_context_t* context, value_t* result) {
    if (!expr || expr->type != EXPR_FUNCTION_CALL) {
        return EXEC_ERROR;
    }
    
    const char* func_name = expr->data.function_call.function_name;
    uint32_t arg_count = expr->data.function_call.arg_count;
    expression_t** args = expr->data.function_call.args;
    
    // Handle aggregate functions
    if (strcmp(func_name, "COUNT") == 0) {
        // COUNT function - for now return 1 as placeholder
        // In real implementation, this would be handled by aggregate executor
        *result = value_make_int64(1);
        return EXEC_SUCCESS;
    } else if (strcmp(func_name, "SUM") == 0 || strcmp(func_name, "AVG") == 0 ||
               strcmp(func_name, "MIN") == 0 || strcmp(func_name, "MAX") == 0) {
        // Other aggregate functions - placeholder
        *result = value_make_int64(0);
        return EXEC_SUCCESS;
    }
    
    // Handle scalar functions
    if (strcmp(func_name, "ABS") == 0 && arg_count == 1) {
        value_t arg_val;
        exec_result_t arg_result = evaluate_expression(args[0], tuple, context, &arg_val);
        if (arg_result != EXEC_SUCCESS) {
            return arg_result;
        }
        
        if (arg_val.is_null) {
            *result = value_make_null();
        } else if (arg_val.type == VALUE_INT32) {
            *result = value_make_int32(abs(arg_val.data.int32_val));
        } else if (arg_val.type == VALUE_INT64) {
            *result = value_make_int64(llabs(arg_val.data.int64_val));
        } else if (arg_val.type == VALUE_DOUBLE) {
            *result = value_make_double(fabs(arg_val.data.double_val));
        } else {
            value_destroy(&arg_val);
            return EXEC_ERROR;
        }
        
        value_destroy(&arg_val);
        return EXEC_SUCCESS;
    }
    
    // Unknown function
    *result = value_make_null();
    return EXEC_ERROR;
}

bool evaluate_qualification(expression_t* qual, tuple_t* tuple, 
                           executor_context_t* context) {
    if (!qual) return true;  // No filter means pass through
    
    value_t result;
    if (evaluate_expression(qual, tuple, context, &result) != EXEC_SUCCESS) {
        return false;
    }
    
    bool passes = !result.is_null && result.type == VALUE_BOOL && result.data.bool_val;
    value_destroy(&result);
    return passes;
}

// Utility functions
const char* exec_result_to_string(exec_result_t result) {
    switch (result) {
        case EXEC_SUCCESS: return "SUCCESS";
        case EXEC_EOF: return "EOF";
        case EXEC_ERROR: return "ERROR";
        case EXEC_QUAL_FALSE: return "QUAL_FALSE";
        case EXEC_RESCAN: return "RESCAN";
        default: return "UNKNOWN";
    }
}

const char* value_type_to_string(value_type_t type) {
    switch (type) {
        case VALUE_NULL: return "NULL";
        case VALUE_BOOL: return "BOOL";
        case VALUE_INT32: return "INT32";
        case VALUE_INT64: return "INT64";
        case VALUE_FLOAT: return "FLOAT";
        case VALUE_DOUBLE: return "DOUBLE";
        case VALUE_STRING: return "STRING";
        case VALUE_BLOB: return "BLOB";
        case VALUE_TIMESTAMP: return "TIMESTAMP";
        default: return "UNKNOWN";
    }
}

void executor_print_stats(executor_context_t* context) {
    if (!context) return;
    
    printf("Executor Statistics:\n");
    printf("  Total tuples processed: %" PRIu64 "\n", context->total_tuples_processed);
    printf("  Total pages read: %" PRIu64 "\n", context->total_pages_read);
    printf("  Total pages written: %" PRIu64 "\n", context->total_pages_written);
    printf("  Execution time: %" PRIu64 " μs\n", context->execution_time_us);
    printf("  Work memory: %u KB\n", context->work_mem_kb);
}

// Error handling
void executor_set_error(executor_context_t* context, int error_code, const char* message) {
    if (!context) return;
    
    context->error_code = error_code;
    
    if (context->error_message) {
        free(context->error_message);
    }
    
    if (message) {
        context->error_message = malloc(strlen(message) + 1);
        if (context->error_message) {
            strcpy(context->error_message, message);
        }
    } else {
        context->error_message = NULL;
    }
}

bool executor_has_error(executor_context_t* context) {
    return context && context->error_code != 0;
}

const char* executor_get_error_message(executor_context_t* context) {
    return (context && context->error_message) ? context->error_message : "Unknown error";
}

// Scan executor implementation
static executor_state_t* create_scan_executor(executor_context_t* context, plan_node_t* plan) {
    executor_state_t* executor = malloc(sizeof(executor_state_t));
    if (!executor) return NULL;
    
    memset(executor, 0, sizeof(executor_state_t));
    
    executor->plan = plan;
    executor->context = context;
    executor->open = scan_open;
    executor->next = scan_next;
    executor->close = scan_close;
    executor->rescan = scan_rescan;
    
    // Create scan-specific state
    scan_executor_state_t* scan_state = malloc(sizeof(scan_executor_state_t));
    if (!scan_state) {
        free(executor);
        return NULL;
    }
    
    memset(scan_state, 0, sizeof(scan_executor_state_t));
    scan_state->table_name = plan->data.scan.table_name;
    scan_state->filter = plan->data.scan.filter;
    
    executor->private_state = scan_state;
    
    return executor;
}

static exec_result_t scan_open(executor_state_t* state) {
    if (!state || state->is_open) return EXEC_ERROR;
    
    scan_executor_state_t* scan_state = (scan_executor_state_t*)state->private_state;
    
    // Create storage iterator (simplified)
    scan_state->iterator = NULL;  // Would create actual storage iterator
    
    state->is_open = true;
    state->tuples_processed = 0;
    state->tuples_returned = 0;
    
    return EXEC_SUCCESS;
}

static exec_result_t scan_next(executor_state_t* state, tuple_t** tuple) {
    if (!state || !state->is_open || !tuple) return EXEC_ERROR;
    
    scan_executor_state_t* scan_state = (scan_executor_state_t*)state->private_state;
    
    // Simplified implementation - would read from storage
    *tuple = NULL;
    scan_state->scanned_tuples++;
    state->tuples_processed++;
    
    return EXEC_EOF;  // No more tuples
}

static void scan_close(executor_state_t* state) {
    if (!state || !state->is_open) return;
    
    scan_executor_state_t* scan_state = (scan_executor_state_t*)state->private_state;
    
    if (scan_state->iterator) {
        // Close storage iterator
        scan_state->iterator = NULL;
    }
    
    state->is_open = false;
}

static void scan_rescan(executor_state_t* state) {
    if (!state) return;
    
    if (state->is_open) {
        scan_close(state);
    }
    
    scan_open(state);
}

// Additional executor implementations would follow similar patterns...
// For brevity, I'm providing the framework and one complete implementation

// Project executor implementation
static executor_state_t* create_project_executor(executor_context_t* context, plan_node_t* plan) {
    executor_state_t* executor = malloc(sizeof(executor_state_t));
    if (!executor) return NULL;
    
    memset(executor, 0, sizeof(executor_state_t));
    
    executor->plan = plan;
    executor->context = context;
    executor->open = project_open;
    executor->next = project_next;
    executor->close = project_close;
    executor->rescan = project_rescan;
    
    // Create child executor
    if (plan->data.project.child) {
        executor->child_count = 1;
        executor->children = malloc(sizeof(executor_state_t*));
        if (!executor->children) {
            free(executor);
            return NULL;
        }
        executor->children[0] = executor_create_tree(context, plan->data.project.child);
        if (!executor->children[0]) {
            free(executor->children);
            free(executor);
            return NULL;
        }
    }
    
    // Create project-specific state
    project_executor_state_t* project_state = malloc(sizeof(project_executor_state_t));
    if (!project_state) {
        if (executor->children) {
            executor_destroy_tree(executor->children[0]);
            free(executor->children);
        }
        free(executor);
        return NULL;
    }
    
    memset(project_state, 0, sizeof(project_executor_state_t));
    project_state->child = executor->children ? executor->children[0] : NULL;
    project_state->expressions = plan->data.project.expressions;
    project_state->column_names = plan->data.project.column_names;
    project_state->expr_count = plan->data.project.expr_count;
    project_state->result_slot = tuple_slot_create(project_state->expr_count);
    
    executor->private_state = project_state;
    
    return executor;
}

static exec_result_t project_open(executor_state_t* state) {
    if (!state || state->is_open) return EXEC_ERROR;
    
    project_executor_state_t* project_state = (project_executor_state_t*)state->private_state;
    
    // Open child if exists
    if (project_state->child) {
        exec_result_t result = executor_open(project_state->child);
        if (result != EXEC_SUCCESS) {
            return result;
        }
    }
    
    state->is_open = true;
    state->tuples_processed = 0;
    state->tuples_returned = 0;
    
    return EXEC_SUCCESS;
}

static exec_result_t project_next(executor_state_t* state, tuple_t** tuple) {
    if (!state || !state->is_open || !tuple) return EXEC_ERROR;
    
    project_executor_state_t* project_state = (project_executor_state_t*)state->private_state;
    
    // Get tuple from child
    tuple_t* child_tuple = NULL;
    if (project_state->child) {
        exec_result_t result = executor_next(project_state->child, &child_tuple);
        if (result != EXEC_SUCCESS) {
            return result;
        }
    }
    
    // Project expressions to create output tuple
    tuple_t* output_tuple = tuple_create(project_state->expr_count);
    if (!output_tuple) {
        return EXEC_ERROR;
    }
    
    // Evaluate projection expressions (simplified)
    for (uint32_t i = 0; i < project_state->expr_count; i++) {
        value_t value;
        if (evaluate_expression(project_state->expressions[i], child_tuple, 
                               state->context, &value) == EXEC_SUCCESS) {
            output_tuple->slot->values[i] = value;
            output_tuple->slot->is_null[i] = value.is_null;
        } else {
            output_tuple->slot->values[i] = value_make_null();
            output_tuple->slot->is_null[i] = true;
        }
    }
    
    *tuple = output_tuple;
    state->tuples_processed++;
    state->tuples_returned++;
    
    return EXEC_SUCCESS;
}

static void project_close(executor_state_t* state) {
    if (!state || !state->is_open) return;
    
    project_executor_state_t* project_state = (project_executor_state_t*)state->private_state;
    
    if (project_state->child) {
        executor_close(project_state->child);
    }
    
    state->is_open = false;
}

static void project_rescan(executor_state_t* state) {
    if (!state) return;
    
    project_executor_state_t* project_state = (project_executor_state_t*)state->private_state;
    
    if (project_state->child) {
        executor_rescan(project_state->child);
    }
    
    state->tuples_processed = 0;
    state->tuples_returned = 0;
}

// Placeholder implementations for other executor types
static executor_state_t* create_index_scan_executor(executor_context_t* context, plan_node_t* plan) {
    // Similar to scan executor but with index access
    return create_scan_executor(context, plan);  // Simplified
}

static executor_state_t* create_filter_executor(executor_context_t* context, plan_node_t* plan) {
    // Would implement filter logic
    return create_scan_executor(context, plan);  // Simplified
}

static executor_state_t* create_join_executor(executor_context_t* context, plan_node_t* plan) {
    if (!context || !plan || plan->type != PLAN_JOIN) {
        return NULL;
    }
    
    // Create base executor state
    executor_state_t* executor = malloc(sizeof(executor_state_t));
    if (!executor) {
        return NULL;
    }
    
    memset(executor, 0, sizeof(executor_state_t));
    executor->plan = plan;
    executor->context = context;
    executor->open = join_open;
    executor->next = join_next;
    executor->close = join_close;
    executor->rescan = join_rescan;
    
    // Set up child executors (left and right)
    executor->child_count = 2;
    executor->children = malloc(2 * sizeof(executor_state_t*));
    if (!executor->children) {
        free(executor);
        return NULL;
    }
    
    executor->children[0] = executor_create_tree(context, plan->data.join.left);
    executor->children[1] = executor_create_tree(context, plan->data.join.right);
    
    if (!executor->children[0] || !executor->children[1]) {
        if (executor->children[0]) executor_destroy_tree(executor->children[0]);
        if (executor->children[1]) executor_destroy_tree(executor->children[1]);
        free(executor->children);
        free(executor);
        return NULL;
    }
    
    // Create join-specific state (nested loop join for now)
    nested_loop_join_state_t* join_state = malloc(sizeof(nested_loop_join_state_t));
    if (!join_state) {
        executor_destroy_tree(executor->children[0]);
        executor_destroy_tree(executor->children[1]);
        free(executor->children);
        free(executor);
        return NULL;
    }
    
    memset(join_state, 0, sizeof(nested_loop_join_state_t));
    join_state->base.left_child = executor->children[0];
    join_state->base.right_child = executor->children[1];
    join_state->base.join_type = plan->data.join.join_type;
    join_state->base.join_condition = plan->data.join.condition;
    join_state->need_new_outer = true;
    
    // Create result tuple slot
    join_state->base.result_slot = tuple_slot_create(10); // Placeholder size
    if (!join_state->base.result_slot) {
        executor_destroy_tree(executor->children[0]);
        executor_destroy_tree(executor->children[1]);
        free(executor->children);
        free(join_state);
        free(executor);
        return NULL;
    }
    
    executor->private_state = join_state;
    
    return executor;
}

static executor_state_t* create_aggregate_executor(executor_context_t* context, plan_node_t* plan) {
    if (!context || !plan || plan->type != PLAN_AGGREGATE) {
        return NULL;
    }
    
    // Create base executor state
    executor_state_t* executor = malloc(sizeof(executor_state_t));
    if (!executor) {
        return NULL;
    }
    
    memset(executor, 0, sizeof(executor_state_t));
    executor->plan = plan;
    executor->context = context;
    executor->open = aggregate_open;
    executor->next = aggregate_next;
    executor->close = aggregate_close;
    executor->rescan = aggregate_rescan;
    
    // Set up child executor
    if (plan->data.aggregate.child) {
        executor->child_count = 1;
        executor->children = malloc(sizeof(executor_state_t*));
        if (!executor->children) {
            free(executor);
            return NULL;
        }
        
        executor->children[0] = executor_create_tree(context, plan->data.aggregate.child);
        if (!executor->children[0]) {
            free(executor->children);
            free(executor);
            return NULL;
        }
    }
    
    // Create aggregate-specific state
    aggregate_executor_state_t* agg_state = malloc(sizeof(aggregate_executor_state_t));
    if (!agg_state) {
        if (executor->children) {
            executor_destroy_tree(executor->children[0]);
            free(executor->children);
        }
        free(executor);
        return NULL;
    }
    
    memset(agg_state, 0, sizeof(aggregate_executor_state_t));
    agg_state->child = executor->children ? executor->children[0] : NULL;
    agg_state->group_by = plan->data.aggregate.group_by;
    agg_state->aggregates = plan->data.aggregate.aggregates;
    agg_state->group_count = plan->data.aggregate.group_count;
    agg_state->agg_count = plan->data.aggregate.agg_count;
    
    // Create hash table for grouping (simple implementation)
    agg_state->hash_table = hash_table_create(64); // 64 buckets
    if (!agg_state->hash_table) {
        if (executor->children) {
            executor_destroy_tree(executor->children[0]);
            free(executor->children);
        }
        free(agg_state);
        free(executor);
        return NULL;
    }
    
    executor->private_state = agg_state;
    
    return executor;
}

static executor_state_t* create_sort_executor(executor_context_t* context, plan_node_t* plan) {
    if (!context || !plan || plan->type != PLAN_SORT) {
        return NULL;
    }
    
    // Create base executor state
    executor_state_t* executor = malloc(sizeof(executor_state_t));
    if (!executor) {
        return NULL;
    }
    
    memset(executor, 0, sizeof(executor_state_t));
    executor->plan = plan;
    executor->context = context;
    executor->open = sort_open;
    executor->next = sort_next;
    executor->close = sort_close;
    executor->rescan = sort_rescan;
    
    // Set up child executor
    if (plan->data.sort.child) {
        executor->child_count = 1;
        executor->children = malloc(sizeof(executor_state_t*));
        if (!executor->children) {
            free(executor);
            return NULL;
        }
        
        executor->children[0] = executor_create_tree(context, plan->data.sort.child);
        if (!executor->children[0]) {
            free(executor->children);
            free(executor);
            return NULL;
        }
    }
    
    // Create sort-specific state
    sort_executor_state_t* sort_state = malloc(sizeof(sort_executor_state_t));
    if (!sort_state) {
        if (executor->children) {
            executor_destroy_tree(executor->children[0]);
            free(executor->children);
        }
        free(executor);
        return NULL;
    }
    
    memset(sort_state, 0, sizeof(sort_executor_state_t));
    sort_state->child = executor->children ? executor->children[0] : NULL;
    sort_state->sort_keys = plan->data.sort.sort_keys;
    sort_state->ascending = plan->data.sort.ascending;
    sort_state->key_count = plan->data.sort.key_count;
    
    // Initialize tuple array with initial capacity
    sort_state->tuple_capacity = 64;
    sort_state->tuples = malloc(sort_state->tuple_capacity * sizeof(tuple_t*));
    if (!sort_state->tuples) {
        if (executor->children) {
            executor_destroy_tree(executor->children[0]);
            free(executor->children);
        }
        free(sort_state);
        free(executor);
        return NULL;
    }
    
    executor->private_state = sort_state;
    
    return executor;
}

static executor_state_t* create_limit_executor(executor_context_t* context, plan_node_t* plan) {
    if (!context || !plan || plan->type != PLAN_LIMIT) {
        return NULL;
    }
    
    // Create base executor state
    executor_state_t* executor = malloc(sizeof(executor_state_t));
    if (!executor) {
        return NULL;
    }
    
    memset(executor, 0, sizeof(executor_state_t));
    executor->context = context;
    executor->plan = plan;
    executor->open = limit_open;
    executor->next = limit_next;
    executor->close = limit_close;
    executor->rescan = limit_rescan;
    
    // Set up child executor
    if (plan->data.limit.child) {
        executor->child_count = 1;
        executor->children = malloc(sizeof(executor_state_t*));
        if (!executor->children) {
            free(executor);
            return NULL;
        }
        
        executor->children[0] = executor_create_tree(context, plan->data.limit.child);
        if (!executor->children[0]) {
            free(executor->children);
            free(executor);
            return NULL;
        }
    }
    
    // Create limit-specific state
    limit_executor_state_t* limit_state = malloc(sizeof(limit_executor_state_t));
    if (!limit_state) {
        if (executor->children) {
            executor_destroy_tree(executor->children[0]);
            free(executor->children);
        }
        free(executor);
        return NULL;
    }
    
    memset(limit_state, 0, sizeof(limit_executor_state_t));
    limit_state->child = executor->children ? executor->children[0] : NULL;
    limit_state->limit = plan->data.limit.limit;
    limit_state->offset = plan->data.limit.offset;
    limit_state->returned_tuples = 0;
    limit_state->skipped_tuples = 0;
    
    executor->private_state = limit_state;
    
    return executor;
}

// Placeholder implementations for the remaining function signatures
static exec_result_t index_scan_open(executor_state_t* state) { return scan_open(state); }
static exec_result_t index_scan_next(executor_state_t* state, tuple_t** tuple) { return scan_next(state, tuple); }
static void index_scan_close(executor_state_t* state) { scan_close(state); }
static void index_scan_rescan(executor_state_t* state) { scan_rescan(state); }

static exec_result_t filter_open(executor_state_t* state) { return scan_open(state); }
static exec_result_t filter_next(executor_state_t* state, tuple_t** tuple) { return scan_next(state, tuple); }
static void filter_close(executor_state_t* state) { scan_close(state); }
static void filter_rescan(executor_state_t* state) { scan_rescan(state); }

// Join executor implementation (nested loop join)
static exec_result_t join_open(executor_state_t* state) {
    if (!state || !state->private_state) {
        return EXEC_ERROR;
    }
    
    nested_loop_join_state_t* join_state = (nested_loop_join_state_t*)state->private_state;
    
    // Open left (outer) child
    if (join_state->base.left_child) {
        exec_result_t result = executor_open(join_state->base.left_child);
        if (result != EXEC_SUCCESS) {
            return result;
        }
    }
    
    // Open right (inner) child
    if (join_state->base.right_child) {
        exec_result_t result = executor_open(join_state->base.right_child);
        if (result != EXEC_SUCCESS) {
            if (join_state->base.left_child) {
                executor_close(join_state->base.left_child);
            }
            return result;
        }
    }
    
    join_state->need_new_outer = true;
    join_state->current_outer_tuple = NULL;
    join_state->inner_scans = 0;
    state->is_open = true;
    
    return EXEC_SUCCESS;
}

static exec_result_t join_next(executor_state_t* state, tuple_t** tuple) {
    if (!state || !state->private_state || !state->is_open) {
        return EXEC_ERROR;
    }
    
    nested_loop_join_state_t* join_state = (nested_loop_join_state_t*)state->private_state;
    
    while (true) {
        // Get new outer tuple if needed
        if (join_state->need_new_outer) {
            if (join_state->current_outer_tuple) {
                tuple_destroy(join_state->current_outer_tuple);
                join_state->current_outer_tuple = NULL;
            }
            
            exec_result_t result = executor_next(join_state->base.left_child, 
                                                &join_state->current_outer_tuple);
            if (result == EXEC_EOF) {
                return EXEC_EOF; // No more outer tuples
            } else if (result != EXEC_SUCCESS) {
                return result;
            }
            
            // Reset inner relation for new outer tuple
            executor_rescan(join_state->base.right_child);
            join_state->need_new_outer = false;
            join_state->inner_scans++;
        }
        
        // Get next inner tuple
        tuple_t* inner_tuple = NULL;
        exec_result_t result = executor_next(join_state->base.right_child, &inner_tuple);
        
        if (result == EXEC_EOF) {
            // No more inner tuples for current outer, get new outer
            join_state->need_new_outer = true;
            continue;
        } else if (result != EXEC_SUCCESS) {
            return result;
        }
        
        // Check join condition
        bool condition_matches = true;
        if (join_state->base.join_condition) {
            condition_matches = join_condition_matches(join_state->current_outer_tuple, 
                                                      inner_tuple, 
                                                      join_state->base.join_condition,
                                                      state->context);
        }
        
        if (condition_matches) {
            // Create joined tuple
            tuple_t* joined_tuple = join_tuples(join_state->current_outer_tuple, 
                                               inner_tuple, 
                                               join_state->base.join_type);
            
            // Clean up inner tuple (we don't own it after join_tuples)
            // Note: join_tuples should handle memory management
            
            if (joined_tuple) {
                *tuple = joined_tuple;
                join_state->base.joined_tuples++;
                state->tuples_returned++;
                return EXEC_SUCCESS;
            }
        }
        
        // Continue to next inner tuple
    }
    
    return EXEC_EOF;
}

static void join_close(executor_state_t* state) {
    if (!state) return;
    
    nested_loop_join_state_t* join_state = (nested_loop_join_state_t*)state->private_state;
    
    if (join_state) {
        // Close child executors
        if (join_state->base.left_child && join_state->base.left_child->is_open) {
            executor_close(join_state->base.left_child);
        }
        if (join_state->base.right_child && join_state->base.right_child->is_open) {
            executor_close(join_state->base.right_child);
        }
        
        // Clean up current outer tuple
        if (join_state->current_outer_tuple) {
            tuple_destroy(join_state->current_outer_tuple);
            join_state->current_outer_tuple = NULL;
        }
        
        join_state->need_new_outer = true;
        join_state->inner_scans = 0;
        join_state->base.joined_tuples = 0;
    }
    
    state->is_open = false;
    state->tuples_returned = 0;
}

static void join_rescan(executor_state_t* state) {
    if (!state) return;
    
    nested_loop_join_state_t* join_state = (nested_loop_join_state_t*)state->private_state;
    
    if (join_state) {
        // Rescan both children
        if (join_state->base.left_child) {
            executor_rescan(join_state->base.left_child);
        }
        if (join_state->base.right_child) {
            executor_rescan(join_state->base.right_child);
        }
        
        // Reset join state
        if (join_state->current_outer_tuple) {
            tuple_destroy(join_state->current_outer_tuple);
            join_state->current_outer_tuple = NULL;
        }
        
        join_state->need_new_outer = true;
        join_state->inner_scans = 0;
        join_state->base.joined_tuples = 0;
    }
    
    state->tuples_returned = 0;
    state->is_open = true;
}

// Sort executor implementation
static exec_result_t sort_open(executor_state_t* state) {
    if (!state || !state->private_state) {
        return EXEC_ERROR;
    }
    
    sort_executor_state_t* sort_state = (sort_executor_state_t*)state->private_state;
    
    // Open child executor
    if (sort_state->child) {
        exec_result_t result = executor_open(sort_state->child);
        if (result != EXEC_SUCCESS) {
            return result;
        }
    }
    
    // Phase 1: Collect all input tuples
    tuple_t* input_tuple;
    sort_state->tuple_count = 0;
    
    while (sort_state->child && 
           executor_next(sort_state->child, &input_tuple) == EXEC_SUCCESS) {
        
        // Expand array if needed
        if (sort_state->tuple_count >= sort_state->tuple_capacity) {
            sort_state->tuple_capacity *= 2;
            tuple_t** new_tuples = realloc(sort_state->tuples, 
                                          sort_state->tuple_capacity * sizeof(tuple_t*));
            if (!new_tuples) {
                return EXEC_ERROR;
            }
            sort_state->tuples = new_tuples;
        }
        
        // Store copy of tuple
        sort_state->tuples[sort_state->tuple_count++] = tuple_copy(input_tuple);
        state->tuples_processed++;
    }
    
    // Close child executor as we've consumed all its output
    if (sort_state->child) {
        executor_close(sort_state->child);
    }
    
    // Phase 2: Sort the tuples
    if (sort_state->tuple_count > 0 && sort_state->sort_keys && sort_state->key_count > 0) {
        sort_tuples(sort_state->tuples, sort_state->tuple_count, 
                   sort_state->sort_keys, sort_state->ascending, sort_state->key_count);
    }
    
    sort_state->tuples_sorted = true;
    sort_state->current_index = 0;
    state->is_open = true;
    
    return EXEC_SUCCESS;
}

static exec_result_t sort_next(executor_state_t* state, tuple_t** tuple) {
    if (!state || !state->private_state || !state->is_open) {
        return EXEC_ERROR;
    }
    
    sort_executor_state_t* sort_state = (sort_executor_state_t*)state->private_state;
    
    if (!sort_state->tuples_sorted) {
        return EXEC_ERROR;
    }
    
    // Return next sorted tuple
    if (sort_state->current_index < sort_state->tuple_count) {
        *tuple = sort_state->tuples[sort_state->current_index++];
        state->tuples_returned++;
        return EXEC_SUCCESS;
    }
    
    // No more tuples
    return EXEC_EOF;
}

static void sort_close(executor_state_t* state) {
    if (!state) return;
    
    sort_executor_state_t* sort_state = (sort_executor_state_t*)state->private_state;
    if (sort_state) {
        if (sort_state->child && sort_state->child->is_open) {
            executor_close(sort_state->child);
        }
        
        // Free all stored tuples
        for (uint32_t i = 0; i < sort_state->tuple_count; i++) {
            if (sort_state->tuples[i]) {
                tuple_destroy(sort_state->tuples[i]);
            }
        }
        
        sort_state->tuple_count = 0;
        sort_state->current_index = 0;
        sort_state->tuples_sorted = false;
    }
    
    state->is_open = false;
    state->tuples_returned = 0;
}

static void sort_rescan(executor_state_t* state) {
    if (!state) return;
    
    sort_executor_state_t* sort_state = (sort_executor_state_t*)state->private_state;
    if (sort_state && sort_state->tuples_sorted) {
        // Just reset index for rescan if already sorted
        sort_state->current_index = 0;
        state->tuples_returned = 0;
        state->is_open = true;
    } else {
        // Full re-sort needed
        sort_close(state);
        sort_open(state);
    }
}

static exec_result_t limit_open(executor_state_t* state) {
    if (!state || !state->private_state) {
        return EXEC_ERROR;
    }
    
    limit_executor_state_t* limit_state = (limit_executor_state_t*)state->private_state;
    
    // Open child executor
    if (limit_state->child) {
        exec_result_t result = executor_open(limit_state->child);
        if (result != EXEC_SUCCESS) {
            return result;
        }
    }
    
    // Reset counters
    limit_state->returned_tuples = 0;
    limit_state->skipped_tuples = 0;
    
    state->is_open = true;
    return EXEC_SUCCESS;
}

static exec_result_t limit_next(executor_state_t* state, tuple_t** tuple) {
    if (!state || !state->private_state || !state->is_open) {
        return EXEC_ERROR;
    }
    
    limit_executor_state_t* limit_state = (limit_executor_state_t*)state->private_state;
    
    // Check if we've returned enough tuples
    if (limit_state->limit > 0 && limit_state->returned_tuples >= limit_state->limit) {
        return EXEC_EOF;
    }
    
    // Get next tuple from child
    tuple_t* child_tuple = NULL;
    while (limit_state->child) {
        exec_result_t result = executor_next(limit_state->child, &child_tuple);
        if (result != EXEC_SUCCESS) {
            return result;
        }
        
        // Skip tuples for OFFSET
        if (limit_state->offset > 0 && limit_state->skipped_tuples < limit_state->offset) {
            limit_state->skipped_tuples++;
            tuple_destroy(child_tuple);
            continue;
        }
        
        // Return this tuple
        *tuple = child_tuple;
        limit_state->returned_tuples++;
        state->tuples_returned++;
        return EXEC_SUCCESS;
    }
    
    return EXEC_EOF;
}

static void limit_close(executor_state_t* state) {
    if (!state) return;
    
    limit_executor_state_t* limit_state = (limit_executor_state_t*)state->private_state;
    if (limit_state && limit_state->child && limit_state->child->is_open) {
        executor_close(limit_state->child);
    }
    
    state->is_open = false;
    state->tuples_returned = 0;
}

static void limit_rescan(executor_state_t* state) {
    if (!state) return;
    
    limit_executor_state_t* limit_state = (limit_executor_state_t*)state->private_state;
    if (limit_state) {
        // Reset counters
        limit_state->returned_tuples = 0;
        limit_state->skipped_tuples = 0;
        
        // Rescan child
        if (limit_state->child) {
            executor_rescan(limit_state->child);
        }
    }
    
    state->tuples_returned = 0;
    state->is_open = true;
}

// Aggregate executor implementation
static exec_result_t aggregate_open(executor_state_t* state) {
    if (!state || !state->private_state) {
        return EXEC_ERROR;
    }
    
    aggregate_executor_state_t* agg_state = (aggregate_executor_state_t*)state->private_state;
    
    // Open child executor
    if (agg_state->child) {
        exec_result_t result = executor_open(agg_state->child);
        if (result != EXEC_SUCCESS) {
            return result;
        }
    }
    
    // Phase 1: Accumulate all input tuples and build aggregations
    tuple_t* input_tuple;
    while (agg_state->child && 
           executor_next(agg_state->child, &input_tuple) == EXEC_SUCCESS) {
        
        // For simple aggregation without GROUP BY
        if (agg_state->group_count == 0) {
            // Global aggregation - just accumulate values
            agg_state->groups_processed = 1; // Mark that we have data
            
            // Accumulate aggregate values for global aggregation
            for (uint32_t i = 0; i < agg_state->agg_count; i++) {
                if (agg_state->aggregates && agg_state->aggregates[i]) {
                    value_t aggregate_val;
                    exec_result_t eval_result = evaluate_expression(agg_state->aggregates[i], 
                                                                  input_tuple, 
                                                                  state->context, 
                                                                  &aggregate_val);
                    
                    if (eval_result == EXEC_SUCCESS) {
                        // For simple aggregation, we'll implement basic accumulation
                        // This could be expanded to support different aggregate functions
                        if (agg_state->aggregates[i]->type == EXPR_FUNCTION_CALL) {
                            const char* func_name = agg_state->aggregates[i]->data.function_call.function_name;
                            
                            // Accumulate COUNT, SUM, etc. (placeholder implementation)
                            if (strcmp(func_name, "COUNT") == 0) {
                                // COUNT implementation will use groups_processed counter
                            } else if (strcmp(func_name, "SUM") == 0) {
                                // SUM would accumulate the values (simplified)
                            }
                        }
                    }
                }
            }
        } else {
            // GROUP BY logic with hash table
            if (agg_state->hash_table && agg_state->group_by) {
                // Calculate hash for the group key
                uint32_t group_hash = hash_tuple(input_tuple, agg_state->group_by, agg_state->group_count);
                
                // Look up existing group in hash table
                uint32_t existing_count = 0;
                tuple_t** existing_tuples = hash_table_lookup(agg_state->hash_table, group_hash, &existing_count);
                
                bool group_exists = false;
                tuple_t* group_tuple = NULL;
                
                // Check if we already have this group key
                if (existing_tuples && existing_count > 0) {
                    for (uint32_t j = 0; j < existing_count; j++) {
                        if (existing_tuples[j]) {
                            // Compare group key values to determine if it's the same group
                            bool keys_match = true;
                            for (uint32_t k = 0; k < agg_state->group_count && keys_match; k++) {
                                value_t current_key, existing_key;
                                
                                if (evaluate_expression(agg_state->group_by[k], input_tuple, 
                                                      state->context, &current_key) == EXEC_SUCCESS &&
                                    evaluate_expression(agg_state->group_by[k], existing_tuples[j], 
                                                      state->context, &existing_key) == EXEC_SUCCESS) {
                                    
                                    // Simple value comparison (could be expanded for different types)
                                    if (current_key.type != existing_key.type ||
                                        (current_key.type == VALUE_INT64 && 
                                         current_key.data.int64_val != existing_key.data.int64_val) ||
                                        (current_key.type == VALUE_STRING && 
                                         strcmp(current_key.data.string_val.data, existing_key.data.string_val.data) != 0)) {
                                        keys_match = false;
                                    }
                                }
                            }
                            
                            if (keys_match) {
                                group_exists = true;
                                group_tuple = existing_tuples[j];
                                break;
                            }
                        }
                    }
                }
                
                if (!group_exists) {
                    // Create new group entry
                    // Create a tuple to represent this group (group keys + aggregate values)
                    tuple_t* new_group_tuple = tuple_create(agg_state->group_count + agg_state->agg_count);
                    if (new_group_tuple) {
                        // Copy group key values
                        for (uint32_t k = 0; k < agg_state->group_count; k++) {
                            value_t group_key_val;
                            if (evaluate_expression(agg_state->group_by[k], input_tuple, 
                                                  state->context, &group_key_val) == EXEC_SUCCESS) {
                                new_group_tuple->slot->values[k] = group_key_val;
                            }
                        }
                        
                        // Initialize aggregate values
                        for (uint32_t i = 0; i < agg_state->agg_count; i++) {
                            if (agg_state->aggregates && agg_state->aggregates[i] &&
                                agg_state->aggregates[i]->type == EXPR_FUNCTION_CALL) {
                                const char* func_name = agg_state->aggregates[i]->data.function_call.function_name;
                                
                                if (strcmp(func_name, "COUNT") == 0) {
                                    new_group_tuple->slot->values[agg_state->group_count + i] = value_make_int64(1);
                                } else if (strcmp(func_name, "SUM") == 0 || strcmp(func_name, "AVG") == 0) {
                                    // Get the value from the current tuple
                                    value_t init_val;
                                    if (evaluate_expression(agg_state->aggregates[i], input_tuple, 
                                                          state->context, &init_val) == EXEC_SUCCESS) {
                                        new_group_tuple->slot->values[agg_state->group_count + i] = init_val;
                                    } else {
                                        new_group_tuple->slot->values[agg_state->group_count + i] = value_make_int64(0);
                                    }
                                }
                            } else {
                                new_group_tuple->slot->values[agg_state->group_count + i] = value_make_int64(0);
                            }
                        }
                        
                        // Insert into hash table
                        hash_table_insert(agg_state->hash_table, new_group_tuple, group_hash);
                        agg_state->groups_processed++;
                    }
                } else if (group_tuple) {
                    // Update existing group's aggregate values
                    for (uint32_t i = 0; i < agg_state->agg_count; i++) {
                        if (agg_state->aggregates && agg_state->aggregates[i] &&
                            agg_state->aggregates[i]->type == EXPR_FUNCTION_CALL) {
                            const char* func_name = agg_state->aggregates[i]->data.function_call.function_name;
                            
                            if (strcmp(func_name, "COUNT") == 0) {
                                // Increment count
                                if (group_tuple->slot->values[agg_state->group_count + i].type == VALUE_INT64) {
                                    group_tuple->slot->values[agg_state->group_count + i].data.int64_val++;
                                }
                            } else if (strcmp(func_name, "SUM") == 0) {
                                // Add to sum
                                value_t current_val;
                                if (evaluate_expression(agg_state->aggregates[i], input_tuple, 
                                                      state->context, &current_val) == EXEC_SUCCESS &&
                                    current_val.type == VALUE_INT64 &&
                                    group_tuple->slot->values[agg_state->group_count + i].type == VALUE_INT64) {
                                    group_tuple->slot->values[agg_state->group_count + i].data.int64_val += 
                                        current_val.data.int64_val;
                                }
                            }
                            // AVG would need additional logic to track count and sum separately
                        }
                    }
                }
            }
        }
        
        state->tuples_processed++;
    }
    
    // Close child executor as we've consumed all its output
    if (agg_state->child) {
        executor_close(agg_state->child);
    }
    
    agg_state->aggregation_done = true;
    state->is_open = true;
    
    return EXEC_SUCCESS;
}

static exec_result_t aggregate_next(executor_state_t* state, tuple_t** tuple) {
    if (!state || !state->private_state || !state->is_open) {
        return EXEC_ERROR;
    }
    
    aggregate_executor_state_t* agg_state = (aggregate_executor_state_t*)state->private_state;
    
    if (!agg_state->aggregation_done) {
        return EXEC_ERROR;
    }
    
    // For simple case - return one result tuple with aggregate values
    if (state->tuples_returned == 0 && agg_state->groups_processed > 0) {
        // Create result tuple
        uint32_t result_columns = agg_state->group_count + agg_state->agg_count;
        tuple_t* result_tuple = tuple_create(result_columns > 0 ? result_columns : 1);
        if (!result_tuple) {
            return EXEC_ERROR;
        }
        
        // Fill in aggregate values (simplified implementation)
        for (uint32_t i = 0; i < agg_state->agg_count; i++) {
            if (agg_state->aggregates && agg_state->aggregates[i] && 
                agg_state->aggregates[i]->type == EXPR_FUNCTION_CALL) {
                
                const char* func_name = agg_state->aggregates[i]->data.function_call.function_name;
                
                if (strcmp(func_name, "COUNT") == 0) {
                    result_tuple->slot->values[i] = value_make_int64(agg_state->groups_processed);
                } else if (strcmp(func_name, "SUM") == 0) {
                    result_tuple->slot->values[i] = value_make_int64(agg_state->groups_processed); // Placeholder
                } else if (strcmp(func_name, "AVG") == 0) {
                    result_tuple->slot->values[i] = value_make_double(1.0); // Placeholder
                } else if (strcmp(func_name, "MIN") == 0 || strcmp(func_name, "MAX") == 0) {
                    result_tuple->slot->values[i] = value_make_int64(1); // Placeholder
                } else {
                    result_tuple->slot->values[i] = value_make_null();
                }
                result_tuple->slot->is_null[i] = false;
            } else {
                result_tuple->slot->values[i] = value_make_null();
                result_tuple->slot->is_null[i] = true;
            }
        }
        
        // Handle case where no aggregates (just COUNT(*))
        if (agg_state->agg_count == 0) {
            result_tuple->slot->values[0] = value_make_int64(agg_state->groups_processed);
            result_tuple->slot->is_null[0] = false;
        }
        
        *tuple = result_tuple;
        state->tuples_returned++;
        return EXEC_SUCCESS;
    }
    
    // No more results
    return EXEC_EOF;
}

static void aggregate_close(executor_state_t* state) {
    if (!state) return;
    
    aggregate_executor_state_t* agg_state = (aggregate_executor_state_t*)state->private_state;
    if (agg_state) {
        if (agg_state->child && agg_state->child->is_open) {
            executor_close(agg_state->child);
        }
        
        // Clean up hash table
        if (agg_state->hash_table) {
            hash_table_destroy(agg_state->hash_table);
            agg_state->hash_table = NULL;
        }
        
        agg_state->aggregation_done = false;
        agg_state->groups_processed = 0;
    }
    
    state->is_open = false;
    state->tuples_returned = 0;
}

static void aggregate_rescan(executor_state_t* state) {
    if (!state) return;
    
    aggregate_close(state);
    aggregate_open(state);
}

// Sorting utility functions

static int tuple_compare_wrapper(const void* a, const void* b, void* context) {
    tuple_t* tuple_a = *(tuple_t**)a;
    tuple_t* tuple_b = *(tuple_t**)b;
    sort_context_t* sort_ctx = (sort_context_t*)context;
    
    // Handle NULL tuples - NULL sorts last
    if (!tuple_a && !tuple_b) return 0;
    if (!tuple_a) return 1;  // tuple_a > tuple_b
    if (!tuple_b) return -1; // tuple_a < tuple_b
    
    for (uint32_t i = 0; i < sort_ctx->key_count; i++) {
        value_t val_a, val_b;
        
        // Evaluate sort key expressions for both tuples
        // Note: We use a minimal context since we don't have full executor context here
        // In a real implementation, this would be passed down from the sort executor
        executor_context_t minimal_context = {0}; // Zero-initialized minimal context
        
        exec_result_t result_a = evaluate_expression(sort_ctx->sort_keys[i], tuple_a, 
                                                    &minimal_context, &val_a);
        exec_result_t result_b = evaluate_expression(sort_ctx->sort_keys[i], tuple_b, 
                                                    &minimal_context, &val_b);
        
        // Handle evaluation errors - treat as NULL
        if (result_a != EXEC_SUCCESS) {
            val_a = value_make_null();
        }
        if (result_b != EXEC_SUCCESS) {
            val_b = value_make_null();
        }
        
        // Handle NULL values in sort keys - NULL sorts last
        if (val_a.is_null && val_b.is_null) {
            value_destroy(&val_a);
            value_destroy(&val_b);
            continue; // Both NULL, try next sort key
        }
        if (val_a.is_null) {
            value_destroy(&val_a);
            value_destroy(&val_b);
            return 1;  // val_a (NULL) > val_b (non-NULL)
        }
        if (val_b.is_null) {
            value_destroy(&val_a);
            value_destroy(&val_b);
            return -1; // val_a (non-NULL) < val_b (NULL)
        }
        
        // Compare the values using the existing value_compare function
        int comparison = value_compare(&val_a, &val_b);
        
        value_destroy(&val_a);
        value_destroy(&val_b);
        
        if (comparison != 0) {
            // Apply sort direction: ascending = normal order, descending = reverse order
            return sort_ctx->ascending[i] ? comparison : -comparison;
        }
        
        // Values are equal, continue to next sort key
    }
    
    // All sort keys are equal
    return 0;
}

void sort_tuples(tuple_t** tuples, uint32_t count, expression_t** sort_keys,
                bool* ascending, uint32_t key_count) {
    if (!tuples || count == 0 || !sort_keys || key_count == 0) {
        return;
    }
    
    sort_context_t sort_ctx = {
        .sort_keys = sort_keys,
        .ascending = ascending,
        .key_count = key_count
    };
    
    // Use manual quicksort implementation for cross-platform compatibility
    // This provides O(n log n) performance with proper expression evaluation
    if (count > 1) {
        quicksort_tuples(tuples, 0, (int)count - 1, &sort_ctx);
    }
}

// Manual quicksort implementation for platforms without qsort_r
static void quicksort_tuples(tuple_t** tuples, int low, int high, sort_context_t* sort_ctx) {
    if (low < high) {
        int pivot_index = partition_tuples(tuples, low, high, sort_ctx);
        quicksort_tuples(tuples, low, pivot_index - 1, sort_ctx);
        quicksort_tuples(tuples, pivot_index + 1, high, sort_ctx);
    }
}

static int partition_tuples(tuple_t** tuples, int low, int high, sort_context_t* sort_ctx) {
    tuple_t* pivot = tuples[high];
    int i = low - 1;
    
    for (int j = low; j < high; j++) {
        if (tuple_compare_wrapper(&tuples[j], &pivot, sort_ctx) <= 0) {
            i++;
            // Swap tuples[i] and tuples[j]
            tuple_t* temp = tuples[i];
            tuples[i] = tuples[j];
            tuples[j] = temp;
        }
    }
    
    // Swap tuples[i+1] and tuples[high] (pivot)
    tuple_t* temp = tuples[i + 1];
    tuples[i + 1] = tuples[high];
    tuples[high] = temp;
    
    return i + 1;
}

// Join utility functions implementation
bool join_condition_matches(tuple_t* left, tuple_t* right, expression_t* condition,
                           executor_context_t* context) {
    if (!condition) {
        return true; // No condition means all tuples match (CROSS JOIN)
    }
    
    // Create a temporary combined tuple for condition evaluation
    tuple_t* combined_tuple = join_tuples(left, right, JOIN_INNER);
    if (!combined_tuple) {
        return false;
    }
    
    // Evaluate the condition against the combined tuple
    value_t result;
    exec_result_t eval_result = evaluate_expression(condition, combined_tuple, context, &result);
    
    // Clean up the combined tuple
    tuple_destroy(combined_tuple);
    
    if (eval_result != EXEC_SUCCESS) {
        return false; // Evaluation error, treat as non-matching
    }
    
    // Handle NULL result
    if (result.is_null) {
        return false;
    }
    
    // Convert result to boolean
    bool matches = false;
    switch (result.type) {
        case VALUE_BOOL:
            matches = result.data.bool_val;
            break;
        case VALUE_INT32:
            matches = result.data.int32_val != 0;
            break;
        case VALUE_INT64:
            matches = result.data.int64_val != 0;
            break;
        case VALUE_FLOAT:
            matches = result.data.float_val != 0.0f;
            break;
        case VALUE_DOUBLE:
            matches = result.data.double_val != 0.0;
            break;
        default:
            matches = false; // Non-boolean types default to false
            break;
    }
    
    value_destroy(&result);
    return matches;
}

tuple_t* join_tuples(tuple_t* left, tuple_t* right, join_type_t join_type) {
    // For INNER JOIN and CROSS JOIN, both tuples must be present
    if (join_type == JOIN_INNER || join_type == JOIN_CROSS) {
        if (!left || !right) {
            return NULL;
        }
    }
    
    // For outer joins, allow NULL tuples and handle appropriately
    // Calculate column counts, treating NULL tuples as having 0 columns
    uint32_t left_cols = (left && left->slot) ? left->slot->column_count : 0;
    uint32_t right_cols = (right && right->slot) ? right->slot->column_count : 0;
    
    // For RIGHT JOIN, if left is NULL, we need to determine left_cols from schema
    // For LEFT JOIN, if right is NULL, we need to determine right_cols from schema
    // For now, we'll use a default of 1 if tuple is NULL but should be present
    if (join_type == JOIN_LEFT_OUTER && !right) {
        right_cols = 1; // Default placeholder - should get from schema
    }
    if (join_type == JOIN_RIGHT_OUTER && !left) {
        left_cols = 1; // Default placeholder - should get from schema
    }
    if (join_type == JOIN_FULL_OUTER) {
        if (!left) left_cols = 1; // Default placeholder
        if (!right) right_cols = 1; // Default placeholder
    }
    
    uint32_t total_cols = left_cols + right_cols;
    
    if (total_cols == 0) {
        return NULL;
    }
    
    // Create new tuple with combined columns
    tuple_t* joined_tuple = tuple_create(total_cols);
    if (!joined_tuple) {
        return NULL;
    }
    
    uint32_t col_idx = 0;
    
    // Handle different join types for column filling
    switch (join_type) {
        case JOIN_INNER:
        case JOIN_CROSS:
            // Both tuples must be present (checked above)
            // Copy left tuple columns
            if (left && left->slot) {
                for (uint32_t i = 0; i < left_cols; i++) {
                    joined_tuple->slot->values[col_idx] = value_copy(&left->slot->values[i]);
                    joined_tuple->slot->is_null[col_idx] = left->slot->is_null[i];
                    col_idx++;
                }
            }
            
            // Copy right tuple columns
            if (right && right->slot) {
                for (uint32_t i = 0; i < right_cols; i++) {
                    joined_tuple->slot->values[col_idx] = value_copy(&right->slot->values[i]);
                    joined_tuple->slot->is_null[col_idx] = right->slot->is_null[i];
                    col_idx++;
                }
            }
            break;
            
        case JOIN_LEFT_OUTER:
            // Left tuple always present, right may be NULL
            // Copy left tuple columns (must exist)
            if (left && left->slot) {
                for (uint32_t i = 0; i < left_cols; i++) {
                    joined_tuple->slot->values[col_idx] = value_copy(&left->slot->values[i]);
                    joined_tuple->slot->is_null[col_idx] = left->slot->is_null[i];
                    col_idx++;
                }
            }
            
            // Handle right tuple - copy if present, otherwise NULL pad
            if (right && right->slot) {
                // Right tuple exists - copy its columns
                for (uint32_t i = 0; i < right_cols; i++) {
                    joined_tuple->slot->values[col_idx] = value_copy(&right->slot->values[i]);
                    joined_tuple->slot->is_null[col_idx] = right->slot->is_null[i];
                    col_idx++;
                }
            } else {
                // Right tuple is NULL - pad with NULL values
                for (uint32_t i = 0; i < right_cols; i++) {
                    joined_tuple->slot->values[col_idx] = value_make_null(); // NULL value for missing tuple
                    joined_tuple->slot->is_null[col_idx] = true; // Mark as NULL
                    col_idx++;
                }
            }
            break;
            
        case JOIN_RIGHT_OUTER:
            // Right tuple always present, left may be NULL
            // Handle left tuple - copy if present, otherwise NULL pad  
            if (left && left->slot) {
                // Left tuple exists - copy its columns
                for (uint32_t i = 0; i < left_cols; i++) {
                    joined_tuple->slot->values[col_idx] = value_copy(&left->slot->values[i]);
                    joined_tuple->slot->is_null[col_idx] = left->slot->is_null[i];
                    col_idx++;
                }
            } else {
                // Left tuple is NULL - pad with NULL values
                for (uint32_t i = 0; i < left_cols; i++) {
                    joined_tuple->slot->values[col_idx] = value_make_null(); // NULL value for missing tuple
                    joined_tuple->slot->is_null[col_idx] = true; // Mark as NULL
                    col_idx++;
                }
            }
            
            // Copy right tuple columns (must exist)
            if (right && right->slot) {
                for (uint32_t i = 0; i < right_cols; i++) {
                    joined_tuple->slot->values[col_idx] = value_copy(&right->slot->values[i]);
                    joined_tuple->slot->is_null[col_idx] = right->slot->is_null[i];
                    col_idx++;
                }
            }
            break;
            
        case JOIN_FULL_OUTER:
            // Either tuple may be NULL - handle both sides with NULL padding
            // Handle left tuple - copy if present, otherwise NULL pad
            if (left && left->slot) {
                // Left tuple exists - copy its columns
                for (uint32_t i = 0; i < left_cols; i++) {
                    joined_tuple->slot->values[col_idx] = value_copy(&left->slot->values[i]);
                    joined_tuple->slot->is_null[col_idx] = left->slot->is_null[i];
                    col_idx++;
                }
            } else {
                // Left tuple is NULL - pad with NULL values
                for (uint32_t i = 0; i < left_cols; i++) {
                    joined_tuple->slot->values[col_idx] = value_make_null(); // NULL value for missing tuple
                    joined_tuple->slot->is_null[col_idx] = true; // Mark as NULL
                    col_idx++;
                }
            }
            
            // Handle right tuple - copy if present, otherwise NULL pad
            if (right && right->slot) {
                // Right tuple exists - copy its columns
                for (uint32_t i = 0; i < right_cols; i++) {
                    joined_tuple->slot->values[col_idx] = value_copy(&right->slot->values[i]);
                    joined_tuple->slot->is_null[col_idx] = right->slot->is_null[i];
                    col_idx++;
                }
            } else {
                // Right tuple is NULL - pad with NULL values
                for (uint32_t i = 0; i < right_cols; i++) {
                    joined_tuple->slot->values[col_idx] = value_make_null(); // NULL value for missing tuple
                    joined_tuple->slot->is_null[col_idx] = true; // Mark as NULL
                    col_idx++;
                }
            }
            break;
            
        default:
            // Unsupported join type
            tuple_destroy(joined_tuple);
            return NULL;
    }
    
    return joined_tuple;
}

// Correlated subquery support functions

// Set a correlation value for a specific column name
void executor_set_correlation_value(executor_context_t* context, const char* column_name, const value_t* value) {
    if (!context || !column_name || !value) {
        return;
    }
    
    // Check if this column name is in the correlation context
    for (uint32_t i = 0; i < context->correlation_count; i++) {
        if (context->correlation_names[i] && strcmp(context->correlation_names[i], column_name) == 0) {
            // Free the old value
            value_destroy(&context->correlation_values[i]);
            // Set the new value
            context->correlation_values[i] = value_copy(value);
            return;
        }
    }
}

// Clear all correlation values
void executor_clear_correlation_values(executor_context_t* context) {
    if (!context) {
        return;
    }
    
    // Clear correlation values but keep the names
    if (context->correlation_values) {
        for (uint32_t i = 0; i < context->correlation_count; i++) {
            value_destroy(&context->correlation_values[i]);
            context->correlation_values[i] = value_make_null();
        }
    }
}